package org.zjt.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import scala.tools.nsc.transform.Constructors;

import java.lang.reflect.Constructor;
import java.util.List;
import java.util.function.Function;

/**
 * DESC 该工具类为了更好的打开资源和释放资源
 *
 * @author
 * @create 2017-04-20 下午9:15,
 **/
public class TestHandler<R>  {
    private static final String MASTER_HOST = "local";
    private static final String APP_NAME = "my_app";
    private static final JavaSparkContext javaSparkContext ;

    static {
        SparkConf sparkConf = new SparkConf().setMaster(MASTER_HOST).setAppName(APP_NAME);
        javaSparkContext = new JavaSparkContext(sparkConf);

        /** 正常退出、异常退出执行该方法，kill线程后不会执行该方法 */
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("程序运行结束(包含正常退出、异常退出)，并释放spark链接");
            javaSparkContext.close();
        }));
    }


    /**
     *  复习：
     *
     *   JDK 8 中提供了一组常用的核心函数接口：
     *
     *       接口	            参数  	返回类型	    描述
     *       Predicate<T>	    T	    boolean	    用于判别一个对象。比如求一个人是否为男性
     *       Consumer<T>	    T	    void	    用于接收一个对象进行处理但没有返回，比如接收一个人并打印他的名字
     *       Function<T, R>	    T	    R	        转换一个对象为不同类型的对象
     *       Supplier<T>	    None	T	        提供一个对象
     *       UnaryOperator<T>	T	    T	        接收对象并返回同类型的对象
     *       BinaryOperator<T>	(T, T)	T	        接收两个同类型的对象，并返回一个原类型对象
     */

    public R executor(Function<JavaSparkContext,R> function) {
        R counts = function.apply(javaSparkContext);
        if ( counts instanceof JavaPairRDD ){
            JavaPairRDD javaRDD = (JavaPairRDD) counts ;
            List<Tuple2<String, Integer>> output = javaRDD.collect();
            for (Tuple2<?,?> tuple : output) {
                System.out.println(tuple._1() + ": " + tuple._2());
            }
        }else if (counts instanceof JavaRDD){
            JavaRDD javaRDD = (JavaRDD) counts ;
            for (Object num : javaRDD.collect()) {
                System.out.println(num);
            }
        }else
            System.out.println(counts);
        javaSparkContext.stop();
        return counts;
    }


    /**
     * 得到基本变量类型方法： (Class<Void>) Class.getPrimitiveClass("void");
     * @param a
     */
    public static void main(String[] a){
        new TestHandler<Void>().executor(
                javaSparkContext -> {
                    try {

                        /**
                         * 破坏Void单利,传输一个Void。
                         */
                        Constructor con = Void.class.getDeclaredConstructor();
                        con.setAccessible(true);
                        Void singetonTest1 = (Void)con.newInstance();
                        return singetonTest1;
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    return null;
                }
        );
    }



}
